/*
 * Copyright (c) 1995, 2013, Oracle and/or its affiliates. All rights reserved.
 * ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 */

package java.io;

/**
 * 管道输入流应该连接到管道输出流;
 * 然后，管道输入流提供写入管道输出流的字节数据。
 * 通常，一个线程从PipedInputStream对象中读取数据，其他线程将数据写入相应的PipedOutputStream中。
 * 不建议在单个线程中同时使用这两个对象，因为它可能导致线程死锁。
 * 管道输入流包含一个缓冲区，在一定范围内解耦读操作和写操作。
 * 如果向连接的管道输出流提供数据字节的线程不再是活动的，那么管道被破坏。
 *
 * @author  James Gosling
 * @see     java.io.PipedOutputStream
 * @since   JDK1.0
 */
public class PipedInputStream extends InputStream {
	// 3个标志，closedByReader是volatile的
    boolean closedByWriter = false;
    volatile boolean closedByReader = false;
    boolean connected = false;

    /* 提醒:读写侧的识别需要更加复杂。
    要么使用线程组(但是线程中的管道呢?)，要么使用终结(但是到下一次GC可能需要很长时间)。
     */
    // 读和写 的线程
    Thread readSide;
    Thread writeSide;

    private static final int DEFAULT_PIPE_SIZE = 1024;

    /**
     * 管道的循环输入缓冲区的默认大小。1024
     * @since   JDK1.1
     */
    // 在允许更改管道大小之前，这是一个常量。为了向后兼容性，将继续维护该字段。
    protected static final int PIPE_SIZE = DEFAULT_PIPE_SIZE;

    /**
     * 用于存放传入数据的循环缓冲区。
     * @since   JDK1.1
     */
    protected byte buffer[];

    /**
     * 从连接管道输出流接收到的下一个数据字节将存储在循环缓冲区中位置的索引。
     * in<0表示缓冲区为空，in==out表示缓冲区已满
     * @since   JDK1.1
     */
    protected int in = -1;

    /**
     * 循环缓冲区中位置的索引，数据的下一个字节将在该位置被这个管道输入流读取。
     * @since   JDK1.1
     */
    protected int out = 0;

    /**
     * 创建一个PipedInputStream，以便它连接到管道输出流src。
     * 写入src的数据字节就可以从这个流中获得输入。
     *
     * @param      src   the stream to connect to.
     * @exception  IOException  if an I/O error occurs.
     */
    public PipedInputStream(PipedOutputStream src) throws IOException {
        this(src, DEFAULT_PIPE_SIZE);
    }

    /**
     * 创建一个PipedInputStream，以便它连接到管道输出流src，并为管道的缓冲区使用指定的管道大小。
     * 写入src的数据字节将作为该流的输入可用。
     *
     * @param      src   the stream to connect to.
     * @param      pipeSize the size of the pipe's buffer.
     * @exception  IOException  if an I/O error occurs.
     * @exception  IllegalArgumentException if {@code pipeSize <= 0}.
     * @since      1.6
     */
    public PipedInputStream(PipedOutputStream src, int pipeSize)
            throws IOException {
         initPipe(pipeSize);
         connect(src);
    }

    /**
     * 创建一个尚未连接的PipedInputStream。
     * 在使用之前，它必须连接到PipedOutputStream。
     */
    public PipedInputStream() {
        initPipe(DEFAULT_PIPE_SIZE);
    }

    /**
     * 创建一个尚未连接的PipedInputStream，并使用指定的管道大小作为管道的缓冲区。
     * 在使用之前，它必须连接到PipedOutputStream。
     *
     * @param      pipeSize the size of the pipe's buffer.
     * @exception  IllegalArgumentException if {@code pipeSize <= 0}.
     * @since      1.6
     */
    public PipedInputStream(int pipeSize) {
        initPipe(pipeSize);
    }

    private void initPipe(int pipeSize) {
         if (pipeSize <= 0) {
            throw new IllegalArgumentException("Pipe Size <= 0");
         }
         // 设置buffer为字节数组
         buffer = new byte[pipeSize];
    }

    /**
     * 使该管道输入流连接到管道输出流src。
     * 如果该对象已经连接到其他管道输出流，则抛出IOException。
     * <p>
     * 如果src是一个未连接的管道输出流，而snk是一个未连接的管道输入流，它们可以通过以下任一调用连接:
     *
     * <pre><code>snk.connect(src)</code> </pre>
     * <p>
     * or the call:
     *
     * <pre><code>src.connect(snk)</code> </pre>
     * <p>
     * The two calls have the same effect.
     *
     * @param      src   The piped output stream to connect to.
     * @exception  IOException  if an I/O error occurs.
     */
    public void connect(PipedOutputStream src) throws IOException {
    	// 调用src的connect方法，其中
    	// snk.in = -1;
        //	snk.out = 0;
        //	snk.connected = true;
        src.connect(this);
    }

    /**
     * 接收一个字节的数据。
     * 如果没有可用的输入，此方法将阻塞。
     * @param b the byte being received
     * @exception IOException If the pipe is <a href="#BROKEN"> <code>broken</code></a>,
     *          {@link #connect(java.io.PipedOutputStream) unconnected},
     *          closed, or if an I/O error occurs.
     * @since     JDK1.1
     */
    protected synchronized void receive(int b) throws IOException {
        checkStateForReceive();
        // PipedOutputStream.write(int b)方法，调用sink.receive(b)
        // 所以调用本方法的是写侧线程
        writeSide = Thread.currentThread();
        // 如果缓冲区已满，进行等待，直到缓冲区空出来
        if (in == out)
            awaitSpace();
        // 如果缓冲区空了，in和out为0
        if (in < 0) {
            in = 0;
            out = 0;
        }
        // 在in位置写入b，同时in++
        buffer[in++] = (byte)(b & 0xFF);
        // 如果in到了末尾，下一次写入in为0（循环缓冲）
        if (in >= buffer.length) {
            in = 0;
        }
    }

    /**
     * 接收数据到字节数组中。
     * 此方法将阻塞，直到有一些输入可用为止。
     * 
     * @param b the buffer into which the data is received
     * @param off the start offset of the data
     * @param len the maximum number of bytes received
     * @exception IOException If the pipe is <a href="#BROKEN"> broken</a>,
     *           {@link #connect(java.io.PipedOutputStream) unconnected},
     *           closed,or if an I/O error occurs.
     */
    synchronized void receive(byte b[], int off, int len)  throws IOException {
        checkStateForReceive();
        writeSide = Thread.currentThread();
        int bytesToTransfer = len;
        // 当bytesToTransfer > 0，进入循环
        while (bytesToTransfer > 0) {
        	// 如果缓冲区已满，进行等待，直到缓冲区空出来
            if (in == out)
                awaitSpace();
            int nextTransferAmount = 0;
            if (out < in) {
            	// 如果顺序为0 out in length，nextTransferAmount为length-in
            	//（复制要先填充这一部分，再填充0 out）
                nextTransferAmount = buffer.length - in;
            } else if (in < out) {
                if (in == -1) {
                	// 如果in为-1，初始化in和out为0，nextTransferAmount为length
                    in = out = 0;
                    nextTransferAmount = buffer.length - in;
                } else {
                	// 否则为out-in
                    nextTransferAmount = out - in;
                }
            }
            if (nextTransferAmount > bytesToTransfer)
                nextTransferAmount = bytesToTransfer;
            assert(nextTransferAmount > 0);
            // 复制到buffer，从in开始，复制nextTransferAmount个字节
            System.arraycopy(b, off, buffer, in, nextTransferAmount);
            bytesToTransfer -= nextTransferAmount;
            off += nextTransferAmount;
            in += nextTransferAmount;
            // in到末尾了，进行循环，in变为0
            if (in >= buffer.length) {
                in = 0;
            }
        }
    }

    private void checkStateForReceive() throws IOException {
    	// 校验connected属性
        if (!connected) {
            throw new IOException("Pipe not connected");
        // 校验closedByWriter，closedByReader属性
        } else if (closedByWriter || closedByReader) {
            throw new IOException("Pipe closed");
        // 校验readSide线程和readSide的状态
        } else if (readSide != null && !readSide.isAlive()) {
            throw new IOException("Read end dead");
        }
    }

    private void awaitSpace() throws IOException {
    	// 如果缓冲区已满，进入循环
        while (in == out) {
            checkStateForReceive();

            /* 缓冲区已满：提醒所有等待的读者*/
            notifyAll();
            try {
            	// 再等待一秒，进行判断in == out，看看之前唤醒的线程是否把缓冲区读取了，把空间空出来
                wait(1000);
            } catch (InterruptedException ex) {
                throw new java.io.InterruptedIOException();
            }
        }
    }

    /**
     * 通知所有等待的线程，数据的最后一个字节已经被接收。
     * PipedOutputStream.close调用这个方法
     */
    synchronized void receivedLast() {
    	// 设置closedByWriter为true，同时通知所有等待的线程
        closedByWriter = true;
        notifyAll();
    }

    /**
     * 从这个管道输入流中读取数据的下一个字节。
     * 值字节作为0到255之间的int值返回。
     * 此方法将一直阻塞，直到输入数据可用、检测到流的末尾或抛出异常为止。
     *
     * @return     the next byte of data, or <code>-1</code> if the end of the
     *             stream is reached.
     * @exception  IOException  if the pipe is
     *           {@link #connect(java.io.PipedOutputStream) unconnected},
     *           <a href="#BROKEN"> <code>broken</code></a>, closed,
     *           or if an I/O error occurs.
     */
    public synchronized int read()  throws IOException {
        if (!connected) {
            throw new IOException("Pipe not connected");
        } else if (closedByReader) {
            throw new IOException("Pipe closed");
        } else if (writeSide != null && !writeSide.isAlive()
                   && !closedByWriter && (in < 0)) {
            throw new IOException("Write end dead");
        }

        // readSide线程为调用read方法的线程
        readSide = Thread.currentThread();
        int trials = 2;
        while (in < 0) {
            if (closedByWriter) {
                /* 如果被writer关闭，返回 EOF */
                return -1;
            }
            // 如果writeSide不生存，报错
            if ((writeSide != null) && (!writeSide.isAlive()) && (--trials < 0)) {
                throw new IOException("Pipe broken");
            }
            /* 可能有一个等待的writer */
            notifyAll();
            try {
                wait(1000);
            } catch (InterruptedException ex) {
                throw new java.io.InterruptedIOException();
            }
        }
        // 返回buffer[out]，然后out++
        int ret = buffer[out++] & 0xFF;
        // out也进行循环
        if (out >= buffer.length) {
            out = 0;
        }
        // 如果读取完了，设置in为-1
        if (in == out) {
            /* now empty */
            in = -1;
        }

        return ret;
    }

    /**
     * 从这个管道输入流中读取最多len字节的数据到字节数组中。
     * 如果到达数据流的末尾，或者如果len超过了管道的缓冲区大小，小于len的字节被读取。
     * 如果len为0，则不读取字节并返回0;
     * 否则，该方法将阻塞，直到至少有1个输入字节可用、检测到流的末尾或抛出异常为止。
     *
     * @param      b     the buffer into which the data is read.
     * @param      off   the start offset in the destination array <code>b</code>
     * @param      len   the maximum number of bytes read.
     * @return     the total number of bytes read into the buffer, or
     *             <code>-1</code> if there is no more data because the end of
     *             the stream has been reached.
     * @exception  NullPointerException If <code>b</code> is <code>null</code>.
     * @exception  IndexOutOfBoundsException If <code>off</code> is negative,
     * <code>len</code> is negative, or <code>len</code> is greater than
     * <code>b.length - off</code>
     * @exception  IOException if the pipe is <a href="#BROKEN"> <code>broken</code></a>,
     *           {@link #connect(java.io.PipedOutputStream) unconnected},
     *           closed, or if an I/O error occurs.
     */
    public synchronized int read(byte b[], int off, int len)  throws IOException {
        if (b == null) {
            throw new NullPointerException();
        } else if (off < 0 || len < 0 || len > b.length - off) {
            throw new IndexOutOfBoundsException();
        } else if (len == 0) {
            return 0;
        }

        /* 可能等待第一个字符 */
        // 先读取一个字符，并可能先等待
        int c = read();
        if (c < 0) {
            return -1;
        }
        b[off] = (byte) c;
        int rlen = 1;
        while ((in >= 0) && (len > 1)) {

            int available;
            // in是receive增加，out是read增加
            // 读取out到in或者到length的内容
            // 0 out in length
            if (in > out) {
                available = Math.min((buffer.length - out), (in - out));
            } else {
            	// 0 in out length
                available = buffer.length - out;
            }

            // 在循环外部预先读取一个字节
            if (available > (len - 1)) {
                available = len - 1;
            }
            System.arraycopy(buffer, out, b, off + rlen, available);
            out += available;
            rlen += available;
            len -= available;

            if (out >= buffer.length) {
                out = 0;
            }
            if (in == out) {
                /* now empty */
                in = -1;
            }
        }
        return rlen;
    }

    /**
     * 返回可以从这个inputstream中读取而不阻塞的字节数。
     *
     * @return the number of bytes that can be read from this input stream
     *         without blocking, or {@code 0} if this input stream has been
     *         closed by invoking its {@link #close()} method, or if the pipe
     *         is {@link #connect(java.io.PipedOutputStream) unconnected}, or
     *          <a href="#BROKEN"> <code>broken</code></a>.
     *
     * @exception  IOException  if an I/O error occurs.
     * @since   JDK1.0.2
     */
    public synchronized int available() throws IOException {
        if(in < 0)
            return 0;
        else if(in == out)
            return buffer.length;
        else if (in > out)
            return in - out;
        else
            return in + buffer.length - out;
    }

    /**
     * 关闭该管道输入流，并释放与该流关联的任何系统资源。
     *
     * @exception  IOException  if an I/O error occurs.
     */
    public void close()  throws IOException {
    	// 设置closedByReader ，in
        closedByReader = true;
        synchronized (this) {
            in = -1;
        }
    }
}
